package com.adingxiong.cft.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.stereotype.Component;

/**
 * @ClassName KafkaConsumerFilter
 * @Description   消息过滤
 * @Author xiongchao
 * @Date 2021/3/24 15:28
 **/
@Component
@Slf4j
public class KafkaConsumerFilter {

    @Autowired
    ConsumerFactory consumerFactory;

    @Bean
    public ConcurrentKafkaListenerContainerFactory filterContainerFactory (){
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        //将被过滤的消息丢弃
        factory.setAckDiscarded(true);
        //消息策略
        factory.setRecordFilterStrategy(consumerRecord ->  {
            if(!consumerRecord.value().toString().equals("zhangsan")){
                return false;
            }
            //返回true 则表示消息被过滤
            return true;
        });
        return factory;
    }


    /**
     * 消息过滤监听
     */

    @KafkaListener(topics = {"topic1"},containerFactory = "filterContainerFactory")
    public void onMessage6(ConsumerRecord<?, ?> record) {
        System.out.println(record.value());
        log.info("消息拦截监听,消费的值:{}",record.value());
    }

}
